package com.hxbd.test.utils;

import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;

public class StopEventPushConsumer {
	
	
	public static final String TOPIC = "meta.Topic";
	public static final String SUBTAG = "subtag";
	public static final String CONSUMERGROUP = "meta.ConsumerGroup";
	public static final String NAMESERVER = "meta.NameServer";
	public static final String OUTPUTFILE = "outputfile";
	public static final String KEY = "KEY";
	private String key;
	private List<String> msgList = new ArrayList<String>();
	public List<String> getMsgList() {
		return msgList;
	}

	private DefaultMQPushConsumer consumer;
	public BufferedWriter bw;
    private int count = 0;
	public  StopEventPushConsumer(Map<String,String> conf) {
		final String consumerGroup = conf.get(CONSUMERGROUP);
		final String nameServer = conf.get(NAMESERVER);
		final String topic = conf.get(TOPIC);
		final String subtag = conf.get(SUBTAG);
		key = conf.get(KEY);
		if (conf.containsKey(OUTPUTFILE)) {
			try {
				bw = new BufferedWriter(new FileWriter(conf.get(OUTPUTFILE),true));
			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}		
		if (topic == null || consumerGroup == null || nameServer == null) {
			throw new IllegalArgumentException(TOPIC + " is " + topic + "\n" + CONSUMERGROUP + " is " + consumerGroup + "\n" + NAMESERVER + " is " + nameServer + "\n");
		}
		consumer = new DefaultMQPushConsumer(consumerGroup);
		setUpRocketMq(nameServer, topic,subtag);
	}
	public static String getDate() {
		 SimpleDateFormat formatter = new SimpleDateFormat("yyyyMMdd_HHmmss");
		 return formatter.format(new Date());
	}
	private void setUpRocketMq(final String nameServer, final String topic,String subtag) {
		System.out.println(String.format("initialize consumer..."+subtag));
		consumer.setInstanceName(subtag);
		consumer.setNamesrvAddr(nameServer);
		consumer.setConsumeMessageBatchMaxSize(10);
		try {
			consumer.subscribe(topic, subtag);
			consumer.registerMessageListener(new MessageListenerConcurrently() {
				@Override
				public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
					try{
						processNextMsgs(msgs);
					}
					catch (Exception ex){
						System.out.println("Exception ex"+ ex);
						return ConsumeConcurrentlyStatus.RECONSUME_LATER;
					}
					return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
				}
			});
			consumer.start();
			try {
				Thread.sleep(10000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		} catch (MQClientException e) {
			System.out.println("MQClientException:----" +e);
		}
	}
	
	public void processNextMsgs(final List<MessageExt> msgs) throws Exception {
	                	for(int i=0; i<msgs.size();i++){
	                		if (key==null){
	                			String msg = new String(msgs.get(i).getBody());
		                		msgList.add(msg);
		                		//获取location数据
		                		if (bw != null) {
		                			try {
		                				bw.write(msg+"\n");
			            			} catch (IOException e) {
			            				// TODO Auto-generated catch block
			            				e.printStackTrace();
			            			}
		                		}	                		
		                		System.out.println(msg);
	                		} 
	                		else if (key!=null && msgs.get(i).getKeys().equals(key)) {
	                			String msg = new String(msgs.get(i).getBody());
		                		msgList.add(msg);
		                		//获取location数据
		                		if (bw != null) {
		                			try {
		                				bw.write(msg+"\n");
			            			} catch (IOException e) {
			            				// TODO Auto-generated catch block
			            				e.printStackTrace();
			            			}
		                		}	                		
		                		System.out.println(msg);
	                		}
	            		} 	            
		
	}

	public void close() {
		try {
			if (null != bw) {
				bw.close();  
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		consumer.shutdown();
		System.out.println("关闭消费者");
	}

}